#include "MsgQServlet.h"
#include "unistd.h"
#include "string.h"
#include <sys/types.h>
#include <sys/msg.h>
#include <sys/ipc.h>
#include <stdlib.h>
#include <errno.h>
#include <thread>
#include "ILogger.h"

#define SYNC_PID  1
#define SYNC_MESSAGE 2
struct MessageQData
{
	long int type;	//消息类型
	union DATA{
        int pid;
        struct {
            char buf[512];
            int size;
        } content;
       
    }data;
};

void createFile(const std::string &path) {
    FILE *hF = fopen(path.c_str(),"w+");
    if (hF) fclose(hF);
}
void MsgQClientServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler)
{
    msgidHandlers[type].push_back(handler);
}
void MsgQClientServlet::request(const void *data,int len)
{
    MessageQData msgQ;
    msgQ.data.pid = getpid();
    msgQ.data.content.size = len;
    memcpy(msgQ.data.content.buf,data,len);
    msgQ.type = SYNC_MESSAGE;
    msgsnd(servermqid, &msgQ, sizeof(MessageQData::data), 0);
}

void MsgQClientServlet::start() {
    //  创建进程队列
    //  获取服务端队列
    key_t serverKey = ftok(_connection.c_str(),123);
    //  send pid 给服务端，服务端这时创建线程专门接受该队列消息
    std::stringstream ss;
    ss << "msgQClient_" <<  getpid();
    createFile(ss.str());
    key_t key = ftok(ss.str().c_str(),123);
  	//获取一个消息队列
    clientmqid = msgget(key, IPC_CREAT | 0664);
    LogI("msgget %d %s", key,strerror(errno));
    struct msqid_ds mqds;
    //获取与消息队列关联的数据结构
    if(msgctl(clientmqid, IPC_STAT, &mqds)!=0){   
       LogE("msgctrl IPC_STAT failed");
       return;
    }
    mqds.msg_qbytes = sizeof(MessageQData::DATA)*10;
    //设置消息队列能够放入的消息数量和
    if(msgctl(clientmqid,IPC_SET, &mqds)!=0){
       LogE("msgctrl IPC_SET failed");
       return;
    }
    LogI("create a client msgq %s:%d",ss.str().c_str(),clientmqid);
    servermqid = msgget(serverKey,IPC_CREAT | 0664);
    MessageQData msgQ;
    msgQ.type = SYNC_PID;
    msgQ.data.pid = getpid();
    msgsnd(servermqid, &msgQ, sizeof(MessageQData::data), 0);
    // 创建一个线程专门接受来自service 的消息
    _client = new std::thread(&MsgQClientServlet::recvthreadLogic,this,clientmqid);
}
void MsgQClientServlet::wait()
{
    ((std::thread*)_client)->join();
}
void MsgQClientServlet::stop()
{
    isRun = false;
    if(msgctl(clientmqid, IPC_RMID, NULL) == -1)
	{
		LogE("msgctl IPC_RMID err");
		return;
	}
}
void MsgQClientServlet::onLoad(INode *node)
{
    INode *child = node->childNode("connection");
    if (child) {
        _connection = child->getStr("name");
    }
    start();
}
void MsgQClientServlet::onUnload() {
    wait();
}
void MsgQClientServlet::onMessage(ISession *session,void* data,int size)
{
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void MsgQClientServlet::onState(SERVLET_STATUS status,ISession *session,int code)
{
    auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
         for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}

class MsgQClientSession : public ISession{
public:
    MsgQClientSession(int mqid) : mqid(mqid)
    {

    }
    void reply(const void *data,int len)
    {
        MessageQData msgQ;

        msgQ.data.pid = getpid();
        msgQ.data.content.size = len;
        memcpy(msgQ.data.content.buf,data,len);
        msgQ.type = SYNC_MESSAGE;
        
        msgsnd(mqid, &msgQ, sizeof(MessageQData::data), 0);
    }
    void close(){

    }
private:
    int mqid;
};

void MsgQClientServlet::recvthreadLogic(int mqid)
{
    while(isRun){
       MessageQData msg;
       if (msgrcv(mqid, &msg, sizeof(msg), 0, 0)==-1) {
           if(errno==ENOMSG){
                LogI("no msg\n");
                continue;
            }else{
                LogE("error while  recv msg %d",errno);
                break;
            }
       } else {
           // handle the msg
           switch (msg.type)
           {
           case SYNC_PID:
               /* code */
               break;
           case SYNC_MESSAGE:
                {
                    MsgQClientSession session(servermqid);
                    onMessage(&session,msg.data.content.buf,msg.data.content.size);
                }break;
           default:
               break;
           }
       }
    }
}


void MsgQServerServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler) {
    msgidHandlers[type].push_back(handler);
}
void MsgQServerServlet::request(const void *data,int len) {
    
}
void MsgQServerServlet::start() {
    //  创建进程队列
    //  获取服务端队列
    createFile(_connection);
    key_t serverKey = ftok(_connection.c_str(),123);
    //  send pid 给服务端，服务端这时创建线程专门接受该队列消息
    servermqid = msgget(serverKey,IPC_CREAT | 0664);

    LogI("msgget %d %s", serverKey,strerror(errno));
    struct msqid_ds mqds;
     //获取与消息队列关联的数据结构
    if(msgctl(servermqid, IPC_STAT, &mqds)!=0){   
       LogE("msgctrl IPC_STAT failed");
       return;
    }
    mqds.msg_qbytes = sizeof(MessageQData::DATA)*10;
    //设置消息队列能够放入的消息数量和
    if(msgctl(servermqid,IPC_SET, &mqds)!=0){
       LogE("msgctrl IPC_SET failed");
       return;
    }
    // 创建一个线程专门接受来自service 的消息
    _thread =  new std::thread(&MsgQServerServlet::recvthreadLogic,this,servermqid);
}
void MsgQServerServlet::wait() {
    ((std::thread*)_thread)->join();
}
void MsgQServerServlet::stop() {
    isRun = false;
    if(msgctl(servermqid, IPC_RMID, NULL) == -1)
	{
		LogE("msgctl IPC_RMID err");
		return;
	}
}
void MsgQServerServlet::onLoad(INode *node) {
    INode *child = node->childNode("connection");
    if (child) {
        _connection = child->getStr("name");
    }
    start();
}
void  MsgQServerServlet::onUnload() {
    wait();
}
void MsgQServerServlet::onMessage(ISession *session,void* data,int size) {
    auto iter = msgidHandlers.find(STREAM);
    if (iter != msgidHandlers.end()) {
        for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,data,size);
    }
}
void MsgQServerServlet::onState(SERVLET_STATUS status,ISession *session,int code) {
    auto iter = msgidHandlers.find(status);
    if (iter != msgidHandlers.end()) {
         for (auto i = iter->second.begin(); i != iter->second.end();++i)
           (*i)(session,NULL,code);
    }
}

class MsgQServerSession : public ISession{
public:
    MsgQServerSession(int mqid) : mqid(mqid)
    {

    }
    void reply(const void *data,int len)
    {
        MessageQData msgQ;

        msgQ.data.pid = getpid();
        msgQ.data.content.size = len;
        memcpy(msgQ.data.content.buf,data,len);
        msgQ.type = SYNC_MESSAGE;
        
        msgsnd(mqid, &msgQ, sizeof(MessageQData::data), 0);
    }
    void close(){

    }
private:
    int mqid;
};

void MsgQServerServlet::recvthreadLogic(int mqid)
{
    while(isRun){
       MessageQData msg;
       if (msgrcv(mqid, &msg, sizeof(msg), 0, 0)==-1) {
           if(errno == ENOMSG){
                LogI("no msg\n");
                continue;
            }else{
                LogE("error while  recv msg %d",errno);
                break;
            }
       } else {
           // handle the msg
           switch (msg.type)
           {
           case SYNC_PID:
               {
                   std::stringstream ss;
                    ss << "msgQClient_" << msg.data.pid;
                    key_t key = ftok(ss.str().c_str(),123);
                    //获取一个消息队列
                    int clientmqid = msgget(key, IPC_CREAT | 0664);
                    // 绑定pid 和 clientmqid
                    pidWithChildIds[msg.data.pid] = clientmqid;
               }
               break;
           case SYNC_MESSAGE:
              {
                  // 构建一个Session 用来返回信息
                  MsgQServerSession session(pidWithChildIds[msg.data.pid]);
                  onMessage(&session,msg.data.content.buf,msg.data.content.size);
              }break;
           default:
               break;
           }
       }
    }
}

REG_PLUGIN(MsgQServerServlet);
REG_PLUGIN(MsgQClientServlet);